package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.event.model.broker.PushTxMsgEvent;
import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.store.TimeWheelModelManager;
import cn.me.alphamq.broker.store.TxMsgManager;
import cn.me.alphamq.broker.util.QueueIdUtil;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.PushTxMsgReq;
import cn.me.alphamq.common.entity.resp.PushTxMsgResp;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.enumeration.LocalTxStateEnum;
import cn.me.alphamq.common.enumeration.TxMsgFlagEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;

/**
 * 推送事务消息的事件的监听器
 *
 * @author f
 */
public class PushTxMsgListener implements Listener<PushTxMsgEvent> {

    @Override
    public void onReceive(PushTxMsgEvent event) throws Exception {
        ChannelHandlerContext ctx = event.getCtx();
        PushTxMsgReq pushTxMsgReq = event.getPushTxMsgReq();
        String msgId = pushTxMsgReq.getMsgId();
        int txFlag = pushTxMsgReq.getTxFlag();
        int txState = pushTxMsgReq.getTransactionState();

        if (TxMsgFlagEnum.HALF_MSG.getCode() == txFlag) {
            TxMsgManager.put(pushTxMsgReq, ctx);
            TimeWheelModelManager.add(DelayMessageModel.newDelayTxMsgModel(msgId));
        } else if (TxMsgFlagEnum.REMAIN_MSG.getCode() == txFlag) {
            if (LocalTxStateEnum.COMMIT.getCode() == txState) {
                TxMsgManager.remove(msgId);
                String topic = pushTxMsgReq.getTopic();
                int queueId = pushTxMsgReq.getQueueId();
                Container.getCommitLogAppendHandler().appendMsg(Message.newTxMsg(
                        topic, QueueIdUtil.getQueueId(topic, queueId), JSON.toJSONBytes(pushTxMsgReq)));
            } else if (LocalTxStateEnum.ROLLBACK.getCode() == txState) {
                TxMsgManager.remove(msgId);
            }
        }

        ctx.writeAndFlush(new TcpMsg(EventCodeEnum.PUSH_TX_MSG_RESP, JSON.toJSONBytes(new PushTxMsgResp(msgId))));
    }
}
